package com.simoniu.flink.scala.batch.transformation

import org.apache.flink.api.scala.ExecutionEnvironment

/**
 * outerJoin：外连接
 * 一共有三种情况
 * 1：leftOuterJoin
 * 2：rightOuterJoin
 * 3：fullOuterJoin
 * Created by simoniu
 */
object BatchOuterJoinScalaDemo {
  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment

    import org.apache.flink.api.scala._
    //初始化第一份数据 Tuple2<用户id,用户姓名>
    val text1 = env.fromCollection(Array((1, "刘强东"), (2, "马斯克"), (3, "马化腾")))
    //初始化第二份数据 Tuple2<用户id,用户所在城市>
    val text2 = env.fromCollection(Array((1, "北京"), (2, "上海"), (4, "广州")))

    //对两份数据集执行leftOuterJoin操作
    text1.leftOuterJoin(text2)
      .where(0)
      .equalTo(0) {
        (first, second) => {
          //注意：second中的元素可能为null
          if (second == null) {
            (first._1, first._2, "null")
          } else {
            (first._1, first._2, second._2)
          }
        }
      }.print()

    println("========================================")

    //对两份数据集执行rightOuterJoin操作
    text1.rightOuterJoin(text2)
      .where(0)
      .equalTo(0) {
        (first, second) => {
          //注意：first中的元素可能为null
          if (first == null) {
            (second._1, "null", second._2)
          } else {
            (first._1, first._2, second._2)
          }
        }
      }.print()


    println("========================================")

    //对两份数据集执行rightOuterJoin操作
    text1.fullOuterJoin(text2)
      .where(0)
      .equalTo(0) {
        (first, second) => {
          //注意：first和second中的元素都有可能为null
          if (first == null) {
            (second._1, "null", second._2)
          } else if (second == null) {
            (first._1, first._2, "null")
          } else {
            (first._1, first._2, second._2)
          }
        }
      }.print()
  }
}
